#==============================================================================
#
#  $Id: dopy.py,v 1.18 2002/03/09 20:09:32 mike Exp $
#
"""
   The DOPY main module.  This houses all of the central classes and base
   classes that comprise the core DOPY system.
"""
#
#  Copyright (C) 1999 Michael A. Muller
#
#  Permission is granted to use, modify and redistribute this code,
#  providing that the following conditions are met:
#
#  1) This copyright/licensing notice must remain intact.
#  2) If the code is modified and redistributed, the modifications must 
#  include documentation indicating that the code has been modified.
#  3) The author(s) of this code must be indemnified and held harmless
#  against any damage resulting from the use of this code.
#
#  This code comes with ABSOLUTELY NO WARRANTEE, not even the implied 
#  warrantee of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  
#
#  $Log: dopy.py,v $
#  Revision 1.18  2002/03/09 20:09:32  mike
#  Added the dopy.init() function.
#
#  Revision 1.17  2002/02/10 20:19:50  mike
#  Added Reactor support.
#
#  Revision 1.16  2001/08/04 21:52:08  mike
#  doc fix.
#
#  Revision 1.15  2001/08/04 21:50:25  mike
#  *** empty log message ***
#
#  Revision 1.14  2001/08/04 19:23:24  mike
#  Added support for multiple threading modes.
#
#  Revision 1.13  2001/07/13 22:36:37  mike
#  Fixed basic holes in threading.
#
#  Revision 1.12  2001/07/13 02:05:49  mike
#  Added two threading modes (mainly so that communication processing doesn't
#  have to consume thread resources).
#
#  Revision 1.11  2001/07/12 23:19:31  mike
#  Added initial support for non-threaded operation.
#
#  Revision 1.10  2001/06/12 00:51:35  mike
#  Converted CRLFs.
#
#  Revision 1.9  1999/07/09 13:48:10  mike
#  Added hooks for dopy_beforeMethod() and dopy_afterMethod()
#
#  Revision 1.8  1999/06/16 14:32:22  mike
#  First cut at name services.
#
#  Revision 1.7  1999/06/14 17:50:15  mike
#  Added the ability to preserve tracebacks across remote method invocations.
#
#  Revision 1.6  1999/06/11 14:03:55  mike
#  Made remote objects transportable.
#
#  Revision 1.5  1999/04/24 21:51:30  mike
#  Bug fixes: context object was never getting released (causing the protocol to
#  never be released).  Switched to traceback module for displaying tracebacks
#  from mainloop CLI.
#
#  Revision 1.4  1999/04/22 00:30:19  mike
#  Added the Context facility to the hub.
#
#
#==============================================================================


"""
   Limitations
   -----------
   
   At this time, output parameters are not supported (except by passing in
   remote objects).
   
   XXX should protocol and protocol info objects be one and the same?
   Or would this only pollute the protocol interface?  There is a speed
   advantage to this - but I think that protocol info may need to manage
   the underlying connection object.
   
"""

try:
   import thread
   _haveThreads = 1
except ImportError:
   _haveThreads = 0

#------------------------------------------------------------------------------
# threading modes
#------------------------------------------------------------------------------

THRD_NONE = 0              # no threading
THRD_SELECT = 1            # run "select()" loop in a background thread
THRD_FUNC = 2              # THRD_SELECT + create new threads for method 
                           # invocations
THRD_COM = 3               # new threads for method communication, and
                           # each Protocol and ServerProtocol
                           

import string, select, sys, traceback

import tb

class Error(Exception):
   """
      Base class for all DOPY errors.
   """
   pass

class InvalidMethod(Error):
   """
      Raised when an invalid method name is cited in a request.
   """
   pass

class InvalidObjectKey(Error):
   """
      Raised when an invalid object key is cited as the target for a 
      request.
   """
   pass

class BadResponseError(Error):
   """
      Raised when a response is received with an ID other than that expected.
   """
   pass

class ProtocolTerminatedError(Error):
   """
      Raised when a protocol is terminated.
   """
   pass

class ImplementedBySubclassError(Error):
   """
      Raised by base class methods that are expected to be implemented
      by derived classes.
   """
   pass

class _Abort(Exception):
   """
      Internal error raise when a failed precondition aborts request
      processing.
   """
   pass

class Reactor:
   """
      This class allows user defined `input/output` systems to be
      plugged into the DOPY Hub's select loop.
      
      This is an abstract base class.
   """
   
   def fileno(self):
      """
         Returns the file number to be used in the select() function
         to check for read, write, and error.  i.e. returns the Reactor's
         "handle".
         
         Must be implemented by derived classes.
      """
      raise NotImplementedError()
   
   def notifyWhenReadable(self):
      """
         This method returns true when the Reactor wants to be notified when
	 its handle is readable.  Will cause the handle to be included in
	 the "read" list of the select().

         Must be implemented by derived classes.
      """
      raise NotImplementedError()
   
   def notifyWhenWritable(self):
      """
         This method returns true when the Reactor wants to be notified
         when its handle is writable.  Will cause the handle to be included
         in the "write" list of the select().
         
         Must be implemented by derived classes.
      """
      raise NotImplementedError()
   
   def notifyOnError(self):
      """
         This method returns true when the Reactor wants to be notified
         when its handle has an error condition.  Will cause the handle to be
         included in the "error" list of the select().
         
         Must be implemented by derived classes.
      """
      raise NotImplementedError()
   
   def handleRead(self):
      """
         This method will be called when the select() loop discovers that the
         reactor's handle is ready to read.
         
         Must be implemented by derived classes.
      """
      raise NotImplementedError()
   
   def handleWrite(self):
      """
         This method will be called when the sslect() loop discovers that the
         reactor's handle is ready to write.
         
         Must be implemented by derived classes.
      """
      raise NotImplementedError()
   
   def handleError(self):
      """
         This method will be called when the sslect() loop discovers that the
         reactor's handle is in an error state.
         
         Must be implemented by derived classes.
      """
      raise NotImplementedError()

class RemoteObject:
   
   """
      The proxy object.
      
      The following public variables exist:
      
      `dopy_protoSpec`::
         An instance of *Protocol* defining the protocol spec for the
         object.  This may also be *None*, indicating that the object
         exists locally.  If it is *None*, there must be a local object
         defined for it.
      `dopy_key`::
         The key for the object.  Should be unique within the protocol
         spec.
      
      Protected variables:
      
      `_dopy_localObject`::
         For a true remote object, this is set to *None*.  If 
         /`dopy_protoSpec`/ is *None*, this contains the object that this
         is a proxy for.         
   """
   
   def __init__(self, protoSpec, key, localObject = None):
      self.dopy_protoSpec = protoSpec
      self.dopy_key = key
      self._dopy_localObject = localObject
      if localObject:
         assert protoSpec is None

   def __getstate__(self):
      dict = self.__dict__.copy()
      if self.dopy_protoSpec:
         dict['dopy_protoSpec'] = self.dopy_protoSpec.getProtocolId()
      
      # local objects _never_ get flattened: on the other side they must have
      # protocols.
      dict['_dopy_localObject'] = None
      
      return dict

   def __setstate__(self, dict):
      self.__dict__ = dict
      
      # if the protocol spec is not none, it is a string identifying the
      # protocol.  If it is none, the _current_ protocol is the one that
      # will be used.
      if self.dopy_protoSpec:
         self.dopy_protoSpec = _hub.getProtocol(self.dopy_protoSpec)
      else:
         self.dopy_protoSpec = _hub.getContext().protocol()
   
class RemoteMethod:
   """
      A remote method stores information about an operation that can
      be performed on a remote object.
   """
   
   def __init__(self, target, name):
      self.target = target
      self.name = name
   
   def __call__(self, *parms, **kwParms):
      req = Request(self.target, self.name, parms, kwParms)
      response = _hub.processRequest(req)
      if response.exception is not None:
         raise response.exception
      else:
         return response.returnValue

class DynamicRemoteObject(RemoteObject):   
   """
      A dynamic remote object has no facility for local validation of
      method invocations: it has no proxy.  Therefore, it is the
      easiest way to execute things remotely.  Attribute errors are very
      costly, though.
   """
   
   def __getattr__(self, attr):
      # allows one of these to be safely pickled
      if attr != '__getinitargs__':
         
         # if the object exists locally, delegate this.
         if self._dopy_localObject:
            return getattr(self._dopy_localObject, attr)
         else:      
            return RemoteMethod(self, attr)
            
      else:
         raise AttributeError()

class RemoteObjectFactory:
   """
      Derived classes create remote objects.
   """
   
   def create(self, protoSpec, key, localObject = None):
      """
         Derived classes must override this function to provide one that
         returns their particular variety of remote object.
         
         parms:
            protoSpec::
               [@Protocol or None] If supplied, this is the protocol for the
	       object.  If *None*, this is an indicator that the object
	       exists locally and a protocol will be provided for it when
	       the remote object is transported.  /localObject/ should not
               be *None* in this case.
            key::
               [string] object key.
            localObject::
               [any]  If not *None*, this is the local (server side) object.
               If this is provided, /protoSpec/ may be *None*.
      """
      raise ImplementedBySubclassError()

class DynamicRemoteObjectFactory(RemoteObjectFactory):
   """
      Creates dynamic remote objects.
   """
   
   def create(self, protoSpec, key, localObject = None):
      return DynamicRemoteObject(protoSpec, key, localObject)

class Request:

   """
      Contains all of the information necessary to invoke a remote method.
      Public variables:
      /id/::
         The message id which is used to associate the request with its
         response (set by the hub before the request is sent).
      /target/::
         The object key that the method is being invoked upon.
      /method/::
         The name of the method.
      /parms/::
         The parameter sequence.
      /kwParms/::
         The keyword parameters.
   """
   
   def __init__(self, target, method, parms, kwParms):
      self.id = None
      self.target = target
      self.method = method
      self.parms = parms
      self.kwParms = kwParms

   def __getstate__(self):
      # replace the object with its key so that we don't get into
      # an infinitely recursive loop looking for __getinitargs__ with 
      # __getattr__
      dict = self.__dict__.copy()
      dict['target'] = self.target.dopy_key
      return dict

class Response:
   """   
      Response to a Request.  Public variables:
      /id/::
         Response id.  This is the same id as that of the matching request.
      /returnValue/::
         The return value of the remote method.
      /exception/::
         The exception raise by the remote method (*None* if no
         exception was raised).
   """
   
   def __init__(self, id, returnValue = None, exception = None):
      self.id = id
      self.returnValue = returnValue
      self.exception = exception

class _AwaitingResponse:
   
   """
      Instances of this class track pending requests.  They are managed by
      the Hub in the sendRequest() method.
      
      Public variables:
      /id/::
         The key that this object is stored under.  The same id as is used in 
         the matching Request and Response objects.
      /response/::
         Allows us to store the response so that it can be returned to the
         caller.
      /threaded/::
         A class level variable that is used to indicate whether we are using
         the threaded model or not.
   """
   
   def __init__(self, id):
      """
         Constructs a new instance using the given id.
      """
      if _hub.isThreaded():
         self.__wait = thread.allocate_lock()
         self.__wait.acquire()
      
      self.id = id
      self.response = None
   
   def responseReceived(self, response):
      """
         Used to indicate that a response has been received and to set the
         response that was received.  Any threads waiting on getResponse()
         will continue.
      """
      self.response = response
      if _hub.isThreaded():
         self.__wait.release()
   
   def getResponse(self):
      """
         Waits for a response to be received and then returns it.
      """
      if _hub.isThreaded():
         self.__wait.acquire()
      else:
         self.response = _hub._getResponse(self.id)
      return self.response

class FileEventSource:
   
   """
      This is any kind of object that can be used within the select() method
      as a source of low-level events.
   """
   
   def fileno(self):
      """
         Derived classes should implement this to provide a method that
         returns a file number for the object, suitable for use in the
         select() function.
      """
      raise ImplementedBySubclassError()

   def isActive(self):
      """
         Returns true for as long as the server is active.
      """
      raise ImplementedBySubclassError()
   
   def shutdown(self):
      """
         Shuts down the connection.
      """
      raise ImplementedBySubclassError()

class ProtocolServer(FileEventSource):
   
   """
      Protocol servers handle incoming connections on a particular protocol.
   """
   
   def acceptConnection(self):
      """
         Returns a new connection for the server.
      """
      raise ImplementedBySubclassError()   

   def isSingleConnection(self):
      """
         Returns true if the server returns a single connection, as in 
         the case of a server for a named pipe or `stdin/stdout`.
         
         Always returns false, derived classes may override.
      """
      return 0
      

class Protocol(FileEventSource):
      
   """
      Handles the nitty-gritty details of sending and recieving requests
      and responses across a communication protocol.
      
      Since protocols must be unpickleable, there may be multiple instances
      of the same protocol object.  Therefore, there should generally be
      some object that the protocol can locate within its __setattr__ method
      that actually houses the single instance of the connection that
      the multiple protocol objects refer to.
   """
   
   def getNext(self):
      """
         Derived classes must implement this to get the next object
         received on the protocol.  Should return *None* if the
         connection closes.
      """
      raise ImplementedBySubclassError()

   def send(self, msg):
      """
         Sends a request or response message through the protocol.
      """
      raise ImplementedBySubclassError()

   def getProtocolId(self):
      """
         Derived classes must use implement this to return a unique protocol
         id.  Protocol ids should be strings beginning with a type
         identifier followed by a colon (e.g. "tcp:") and followed by a value
         that uniquely identifies the connection.
      """
      raise ImplementedBySubclassError()      

class ProtocolInfo:

   """
      Protocol info manages the current message id and a list of requests.
      Public variables:
      /requests/::
         A dictionary mapping message ids to request locks.
      /proto/::
         The Protocol object itself.
   """
      
   def __init__(self, protocol):
      """
         Constructs a ProtocolInfo object from the Protocol itself.
      """
      self.__lk = thread.allocate_lock()
      self.__counter = 0
      self.requests = {}
      self.proto = protocol
   
   def getNextId(self):
      """
         Returns the next available message id.
      """
      self.__lk.acquire()
      rc = self.__counter
      self.__counter = self.__counter + 1
      self.__lk.release()
      return rc

class Context:
   """
      Instances of this class manage information about the caller context
      which may be used during the processing of a remote method.
   """

   def __init__(self, protocol):
      self.__protocol = protocol
   
   def protocol(self):
      """
         Returns the protocol which was used to communicate to the hub.
      """
      return self.__protocol

class Hub:
   
   """
      The Hub is the equivalent of the ORB - a central context for everything
      in the distributed object system.
      
      The Hub has a *registry* which can be as simple as a dictionary
      or as sophisticated as a CORBA NamingContext.  It must provide
      the __getitem__ and __setitem__ methods to dereference and assign
      keys to objects.
   """
   
   def __init__(self, registry, 
                threadMode = _haveThreads and THRD_FUNC or THRD_NONE
                ):
      self.__registry = registry
      self.__server = None
      self.__threadMode = threadMode
      self.__protocolInfo = {}
      self.__reactors = []
      
      if self.isThreaded():
         self.__context = {}
         
         # stuff to track the status of the select thread (in 
         # THRD_SELECT and THRD_FUNC modes)
         self.__selectThreadStarted = 0
         self.__selectThreadStartLock = thread.allocate_lock()
         self.__selectLoopProcessing = thread.allocate_lock()
         
         # used to keep track of the number of server threads in THRD_COM
         # mode.
         self.__serverThreads = 0
         
      else:
         self.__context = None

   def isThreaded(self):
      """
         Returns true if the hub is threaded, false if not.  The hub
         is threaded in all threads modes except *`THRD_NONE`*
      """
      return self.__threadMode > THRD_NONE
   
   def getThreadMode(self):
      """
         Returns the hub's thread mode.  Thread modes are:

         *`THRD_NONE`* (0)::
            no threading
         *`THRD_SELECT`* (1)::
            run "select()" loop in a background thread
         *`THRD_FUNC`* (2)::
            Like *`THRD_SELECT`* but also create new threads for method 
            invocations
         *`THRD_COM`* (3):: 
            Create a listening thread for each Protocol and ServerProtocol
	    and each method invocation.
      """
      return self.__threadMode
   
   def addServer(self, server):
      """
         Registers /server/ (an instance of ProtocolServer).  In threaded
	 communication mode, the hub will begin handling connections on the
	 server in another thread.
      """
      if server.isSingleConnection():
         self.addProtocol(server.acceptConnection())
      else:
         self.__server = server
         if self.__threadMode == THRD_COM:
            thread.start_new_thread(self.__serviceConnections, (server,))
         else:
            self.__needSelectThread()

   def addObject(self, key, object):
      """
         Registers the /object/ as a remote object under the name /key/ (a 
         string).  Any kind of object can be registered.
         
         Note that a key consisting of name parts seperated by forward
         slashes can be used, but it does not produce the results that one
         might expect after reading the documentation on @getObject() - i.e.
         the object is not stored in a nested context.
      """
      self.__registry[key] = object

   def removeObject(self, key):
      """
         Deregisters the object listed under the given /key/.
         
         parms:
            key::
               [string] object key.
      """
      del self.__registry[key]

   def __getObject(self, node, splitKey):
      """
         Private version of @getObject(), in which /splitKey/ is a list.
      """
      node = node[splitKey[0]]
      remainder = splitKey[1:]
      if remainder:
         return self.__getObject(node, remainder)
      else:
         return node

   def getObject(self, fullKey):
      """
         Returns the object (the object that was added using #addObject()#,
         not necessarily a @RemoteObject) associated with /fullKey/.
         
         /fullKey/ can resemble a UNIX path - that is, it can consist
         of name components seperated by forward slashes.  If it is, 
         and there is nothing matching the key in the registry, 
         a recursive retrieval is performed.
      """
      splitKey = string.split(fullKey, '/')
      if not self.__registry.has_key(fullKey):
         return self.__getObject(self.__registry, splitKey)
      else:
         return self.__registry[fullKey]

   def makeAnonObject(self, object, remoteFactory = None):
      """
         Makes /object/ into an anonymous remote object.  Returns an instance 
         of a @RemoteObject derivative for the object.
         
         Anonymous remote objects are remote objects that are registered
         without any particular key - presumably for temporary use.  Actually,
         the key is the object's *id* - the address of the object.
         
         If /remoteFactory/ is supplied, it should be an instance of a class
         derived from @RemoteObjectFactory.  If it is not supplied, 
         @DynamicRemoteObjectFactory is used to create and return a 
         @DynamicRemoteObject.
      """
      key = `id(object)`
      self.addObject(key, object)
      if not remoteFactory:
         remoteFactory = DynamicRemoteObjectFactory()
      return remoteFactory.create(None, key, object)

   def makeRemoteObject(self, objectId, remoteFactory = None):
      """
         Returns a remote object for the given object id.  An object id
         consists of a protocol id followed by a colon and the object key.
         
         If /remoteFactory/ is supplied, it should be an instance of a class
         derived from @RemoteObjectFactory.  If it is not supplied, 
         @DynamicRemoteObjectFactory is used to create and return a 
         @DynamicRemoteObject.
      """
      # break the object id up into the protocol id and the object key
      parsedId = string.split(objectId, ':')
      key = parsedId[-1]
      protocolId = string.join(parsedId[:-1], ':')
      
      if not remoteFactory:
         remoteFactory = DynamicRemoteObjectFactory()
      return remoteFactory.create(self.getProtocol(protocolId), key)

   def getRemoteObject(self, key, remoteFactory = None):
      """
         Returns a remote object for an object that has been added to the
         current hub identified by /key/.
         
         parms:
            key::
               [string] object key
            remoteFactory::
               [@RemoteObjectFactory or None] if present, this is the
               remote object factory used to create the remote object.
               If not, @DynamicRemoteObjectFactory is used.
      """
      
      if not remoteFactory:
         remoteFactory = DynamicRemoteObjectFactory()
      return remoteFactory.create(None, key)

   def addProtocol(self, proto):
      """
         This function should generally be called by the protocol constructor
         to register itself with the hub.
      """
      self.__protocolInfo[proto.getProtocolId()] = ProtocolInfo(proto)
      if self.__threadMode == THRD_COM:
         thread.start_new_thread(self.__serviceClient, (proto,))
      else:
         self.__needSelectThread()

   def getProtocol(self, protoId):
      """
         This is used to get or create a protocol from its id.  Returns the
         protocol if it already exists, creates it if it doesn't.
      """
      try:
         return self.__protocolInfo[protoId].proto
      except KeyError, ex:
         global protocolFactory
         return protocolFactory.create(protoId)

   def getContext(self):
      """
         Returns the current context object.  This is only available from
         within a remote method call.  If it is called from somewhere
         other than a remotely invoked method, *None* is returned.
      """
      if self.isThreaded():
         return self.__context.get(thread.get_ident())
      else:
         return self.__context

   def addReactor(self, reactor):
      """
         Adds a new reactor object to the hub's internal list.
         This will cause the reactor to be included in the select loop.
         
         parms:
            reactor::
               [@Reactor]
      """
      self.__reactors.append(reactor)

   def processRequest(self, request):
      """
         Processes a request (an instance of @Request) and returns
         the response (a @Response).
      """
      # get the protocol and protocol info objects
      proto = request.target.dopy_protoSpec
      protoInfo = self.__protocolInfo[proto.getProtocolId()]
      
      # give the request a unique id
      request.id = protoInfo.getNextId()
      
      # build the object that we will use to get our response
      resp = _AwaitingResponse(request.id)
      protoInfo.requests[request.id] = resp
      
      # send the request
      proto.send(request)
      
      # get and return the response
      return resp.getResponse()

   def __serviceConnections(self, server):
      """
         This is executed in a thread to begin servicing client connections
         for a particular server.
      """
      # get a lock to protect critical section
      self.__selectThreadStartLock.acquire()
      try:
         # if there are no other server threads, acquire the lock
         if not self.__serverThreads:
            self.__selectLoopProcessing.acquire()
         self.__serverThreads = self.__serverThreads + 1
      finally:
         self.__selectThreadStartLock.release()
         
      while server.isActive():
         server.acceptConnection()

      # lock again
      self.__selectThreadStartLock.acquire()
      try:
         self.__serverThreads = self.__serverThreads - 1
         
         # last one out releases the lock so that we can terminate the
         # mainloop
         if not self.__serverThreads:
            self.__selectLoopProcessing.release()
      finally:
         self.__selectThreadStartLock.release()

   def __connectionClosed(self, protoId, protoInfo):
      """
         Called when a connection is closed.
      """
      for waiter in protoInfo.requests.values():
         resp = Response(waiter.id, 
                         exception = ProtocolTerminatedError(protoId)
                         )
         waiter.responseReceived(resp)
      del self.__protocolInfo[protoId]
   
   def __serviceClient(self, clientConnection):
      myid = clientConnection.getProtocolId()
      proto = self.__protocolInfo[myid]
      
      # context will be associated with this thread.
      self.__setContext(Context(clientConnection))
      
      while clientConnection.isActive():      
         request = clientConnection.getNext()
         
         # if the getNext() method returned None, the connection has gone
         # away: clean up and exit
         if not request:
            self.__connectionClosed(myid, proto)
            break
            
         elif isinstance(request, Response):
            proto.requests[request.id].responseReceived(request)
         else:
            thread.start_new_thread(self.__serviceFunction, 
                                    (clientConnection, request)
                                    )
            
      self.__clearContext()

   def __serviceFunction(self, clientConnection, request):
      self.__setContext(Context(clientConnection))
      response = self.__processRequest(request, clientConnection)
      clientConnection.send(response)      
      self.__clearContext()
   
   def __processRequest(self, request, protocol):
      try:
         try:
            obj = self.getObject(request.target)
         except Exception, ex:
            # check for all kinds of exceptions because an attempt to resolve
            # an object could end up going into user code.
            response = Response(request.id, exception = InvalidObjectKey(ex))
            raise _Abort()
            
         try:
            meth = getattr(obj, request.method)
         except AttributeError, ex:
            response = Response(request.id, exception = InvalidMethod(ex))
            raise _Abort()
         
         try:
            # call the before function if it exists
            if hasattr(obj, 'dopy_beforeMethod'):
               obj.dopy_beforeMethod(request)
               
            # call the method and create the response message
            retval = apply(meth, request.parms, request.kwParms)
            response = Response(request.id, returnValue = retval)
                           
         except Exception, ex:         
            tb.saveTraceback(ex)
            response = Response(request.id, exception = ex)
         
         try:
         
            # call the after function if it exists
            if hasattr(obj, 'dopy_afterMethod'):
               obj.dopy_afterMethod(request, response)
         except:
            pass
            
      except _Abort, ex:
         pass
      return response      

   def __getAllProtocols(self):
      """
         Internal method to return all @Protocol objects as a list.
      """
      prots = []
      for p in self.__protocolInfo.values():
         prots.append(p.proto)
      return prots

   def __getSelectors(self, servers, rd, wr, er):
      """
         Creates lists of objects for the select lists.
         
         parms:
            servers::
               [list<ProtocolServer>] list of servers
            
      """
      rd[:] = er[:] = servers + self.__getAllProtocols()
      wr[:] = []

      # add all of the reactors to the set
      for reactor in self.__reactors:
         if reactor.notifyWhenReadable():
            rd.append(reactor)
         if reactor.notifyWhenWritable():
            wr.append(reactor)
         if reactor.notifyOnError():
            er.append(reactor)

   def __selectLoop(self, msgId = None):
   
      # create a basic list of servers that we are going to listen to
      if self.__server:
         servers = [self.__server]
      else:
         servers = []

      # create the initial list of things we're interested in reading from
      rd = []
      wr = []
      er = []
      self.__getSelectors(servers, rd, wr, er)      
      
      while rd:
         rdx, wrx, erx = select.select(rd, wr, er)
         if erx:
            for erxi in erx:
               if isinstance(erxi, Reactor):
                  erxi.handleError()
               else:
                  print 'error %s'
         while rdx:
            s = rdx.pop()
            if isinstance(s, ProtocolServer):
               self.addProtocol(s.acceptConnection())
            elif isinstance(s, Protocol):
               self.__setContext(Context(s))
               req = s.getNext()
               if isinstance(req, Response):
                  if msgId is None:
                     protocolInfo = self.__protocolInfo[s.getProtocolId()]
                     protocolInfo.requests[req.id].responseReceived(req)
                  elif req.id != msgId:
                     raise BadResponseError("message id %s received, %s "
                                             "expected." % msgId)
                  else:
                     return req
               elif isinstance(req, Request):
                  if self.__threadMode >= THRD_FUNC:
                     thread.start_new_thread(self.__serviceFunction, (s, req))
                  else:
                     resp = self.__processRequest(req, s)
                     s.send(resp)            
               elif req is None:
                  # remove the protocol
                  del self.__protocolInfo[s.getProtocolId()]
               else:
                  # XXX we got a bad object over the pipe.  Close down
                  # the connection.
                  rd.remove(s)
                  
               self.__clearContext()
            else:
               # must be a reactor
               s.handleRead()
         
         # handle writable objects
         for s in wrx:
            if isinstance(s, Reactor):
               s.handleWrite()
            else:
               assert 0, "Non-reactor found in writable list"

         # rebuild the list of objects to check in select()
         self.__getSelectors(servers, rd, wr, er)
      
      # notify anyone who cares that we're done
      if self.__threadMode != THRD_NONE:
         self.__selectLoopProcessing.release()

   def __setContext(self, context):
      if self.isThreaded():
         self.__context[thread.get_ident()] = context
      else:
         self.__context = context
   
   def __clearContext(self):
      if self.isThreaded():
         del self.__context[thread.get_ident()]
      else:
         self.__context = None
   
   def __execInputLine(self):
      try:
         inp = raw_input('#> ')
      except EOFError:
         return 1
      if inp[:2] == '@q':
         return 1
      try:
         exec inp
      except SystemExit, ex:
         raise
      except Exception, ex:
         traceback.print_exc()
      
      return 0

   def __needSelectThread(self):
   
      # make sure that we are threaded
      if self.__threadMode == THRD_NONE:
         return
         
      self.__selectThreadStartLock.acquire()
      try:
         if not self.__selectThreadStarted:
            self.__selectLoopProcessing.acquire()
            thread.start_new_thread(self.__selectLoop, ())
            self.__selectThreadStarted = 1
      finally:
         self.__selectThreadStartLock.release()
   
   def mainloop(self):
      if self.__threadMode != THRD_NONE:
         # wait for the select loop to finish processing
         self.__selectLoopProcessing.acquire()
         self.__selectLoopProcessing.release()
      else:
         self.__selectLoop()

   def _getResponse(self, responseId):
      """
         This function reenters the message processing loop and waits for a
	 particular response.

         If any response other that response should be received, a 
         @BadResponse exception will be raised.
      """
      assert self.__threadMode != THRD_COM
      return self.__selectLoop(responseId)

class ProtocolFactory:
   """
      The protocol factory is a singleton that is always in existence - it
      manages the creation functions for different protocols.
      
      Protocols should each provide a factory method that creates the protocol
      from a list of strings - these being the fields (as divided by colons)
      of the protocol id.
   """

   def __init__(self):
      self.protos = {}
   
   def register(self, protoTypeId, protoFact):
      self.protos[protoTypeId] = protoFact
   
   def create(self, protoId):
      name = string.split(protoId, ':')
      return self.protos[name[0]](name)

protocolFactory = ProtocolFactory()

_hub = None

def init(argv = None, threadMode = _haveThreads and THRD_FUNC or THRD_NONE):
   """
      This function allows you to explicitly initialize the DOPY hub.
      It can be used to pass in options from the command line and
      set defaults from within your program.  It is not necessary to call
      this, but if you choose to use this, you must do so prior to calling
      _any other DOPY function_.

      If /argv/ is not None, returns a list of all unknown command line
      arguments passed in in /argv/ (excluding argv[0], which is the program
      name).  Otherwise returns *None*.

      parms:
         argv::
            [list<string> or None] If provided, this should be the value
            of #sys.argv#.  This parameter allows DOPY to interpret its
            own command line arguments.  Unknown arguments are returned
            in a list to the caller.
         threadMode::
            Indicates the threading mode.
   """
   global _hub
   
   # make sure that we are not too late
   if _hub:
      raise RuntimeError("DOPY hub has already been initialized")
   
   # make sure that the user didn't override to a thread mode that isn't
   # accessible
   if not _haveThreads and threadMode != THRD_NONE:
      raise RuntimeError('Your python interpreter does not have thread '
                          'support'
                         )
   

   # parse the arguments, overriding any values passed in and constructing
   # the list of unused arguments

   if argv:   
      args = argv[1:]
      unused = []
      last = ''
      while args:
         arg = args.pop(0)
         if last == '-dopy-thread':
            if arg == 'none':
               threadMode = THRD_NONE
            elif arg == 'select':
               threadMode = THRD_SELECT
            elif arg == 'func':
               threadMode = THRD_FUNC
            elif arg == 'com':
               threadMode = THRD_COM
            else:
               raise ValueError('Invalid thread mode %s' % repr(arg))
            last = ''
            continue
         
         if arg == '-dopy-thread':
            last = arg
         else:
            unused.append(arg)
   else:
      unused = None
   
   # create the hub singleton
   _hub = Hub({}, threadMode)
   
   # return unused arguments
   return unused

def getHub():
   global _hub

   if not _hub:
      init()
   return _hub
